package io.kimmking.common.zookeeper.dlock;

import cn.hutool.core.util.ObjectUtil;
import io.kimmking.common.zookeeper.properties.ZookeeperProperty;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Slf4j
public class NonFairZKDistributedLock extends AbstractZKDistributedLock{

    public NonFairZKDistributedLock(final ZookeeperProperty property){
        super(property);
    }

    /**
     * 此处的leaseTime释放锁的时间没使用到
     * @param key
     * @param leaseTime
     * @param unit
     * @throws Exception
     */
    @Override
    protected boolean acquire(String key, long leaseTime, TimeUnit unit) throws Exception{
        key = getLockPath(key);
        while (true) {
            try {
                zk.create(key, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                log.info("{} get the ZK lock", Thread.currentThread().getName());
                return true;
            } catch (KeeperException.NodeExistsException e) {
                //说明锁被占用，注册监听器等待占用的节点被释放
                CountDownLatch latch = new CountDownLatch(1);
                Stat stat = zk.exists(key, new LockWatcher(latch));
                if (!ObjectUtil.isNull(stat)) {
                    log.info("{} waitting for {} released lock", Thread.currentThread().getName(), key);
                    latch.await();
                    latch = null;
                }
            } catch (KeeperException | InterruptedException ignored) {
                return false;
            }
        }
    }

    @Override
    protected boolean tryAcquire(String key, long waitTime, long leaseTime, TimeUnit unit) throws Exception{
        if(waitTime < 1L) {
            log.error("{} should be greater than 0L", "waitTime");
            return false;
        }
        String combineKey = getLockPath(key);
        long begin_time = System.currentTimeMillis();
        long current_time;
        while (true) {
            try {
                current_time = System.currentTimeMillis();
                if(current_time - begin_time > waitTime) {
                    log.warn("get lock timeout!");
                    return false;
                }
                zk.create(combineKey, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                log.info("{} get the ZK lock", Thread.currentThread().getName());

                return true;
            } catch (KeeperException.NodeExistsException e) {
                //说明锁被占用，注册监听器等待占用的节点被释放
                CountDownLatch latch = new CountDownLatch(1);
                Stat stat = zk.exists(key, new LockWatcher(latch));
                if (!ObjectUtil.isNull(stat)) {
                    log.info("{} waitting for {} released lock", Thread.currentThread().getName(), combineKey);
                    latch.await();
                    latch = null;
                }
            } catch (KeeperException | InterruptedException ignored) {
                return false;
            }
        }
    }

    @Override
    protected boolean unAcquire(final String key) {
        try {
            String rootKey = root + "/" + Objects.requireNonNull(key, "lockKey can not be NULL!");
            log.info("{} will unlock", Thread.currentThread().getName());
            //删除zk节点
            zk.delete(Objects.requireNonNull(rootKey, "lockName can not be null"), -1);
            return true;
        } catch (InterruptedException | KeeperException e) {
            return false;
        }
    }


    private String getLockPath(String key){
        return root + "/" + Objects.requireNonNull(key, "lockKey can not be NULL!");
    }
    /**
     * 用于监听ZK节点被删除的事件
     */
    private static class LockWatcher implements Watcher {

        private final CountDownLatch latch;

        public LockWatcher (CountDownLatch countDownLatch) {
            this.latch = countDownLatch;
        }

        @Override
        public void process(WatchedEvent watchedEvent) {
            if (watchedEvent.getType() == Event.EventType.NodeDeleted)
                latch.countDown();
        }
    }
}